/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.nio.ByteBuffer;
import java.util.Comparator;
import java.util.logging.Logger;

import edu.ucsd.db.bassarisk.data.api.IValueType;
import edu.ucsd.db.bassarisk.data.api.IValueType.ITypeTools;
import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.storage.common.IColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.loading.ColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.reading.IDrawerIndexReader.CardInfo;

/**
 * An intermediate level class in our multi-teared column store implementation.
 * 
 * In the metaphor, this class corresponds to a card-catalog drawer, full of
 * posting cards. The drawer is a single block of data in the underlying file,
 * and each of its cards is an independently compressed segment of that block.
 * 
 * The DrawerReader class, with the help of the DrawerHeader, manages the cards
 * in the drawer, understands the card layout within the drawer, and most
 * importantly, directs various types of reads (random and sequential) to the
 * correct cards.
 * 
 * @author nbales
 */
public class DrawerReader<VT extends IValueType> implements IColumnSegmentReader<VT> {

  private final CardInfo<VT> cardInfo;
  private final ICardReader<VT> dataCardReader;
  private IColumnSegmentInfo<VT> drawerInfo;
  private final IDrawerIndexReader<VT> indexReader;
  
  private boolean initialized;

  protected Comparator<IValueReference<VT>> valueComparator;
  
  private static final Logger LOG = 
      Logger.getLogger(DrawerReader.class.getName());

  public DrawerReader(final ICardReader<VT> dataCardReader,
                      final IDrawerIndexReader<VT> indexReader,
                      final ITypeTools<VT> columnTypeTools) {
    this.dataCardReader = dataCardReader;
    this.indexReader = indexReader;
    this.cardInfo = new CardInfo<VT>(columnTypeTools);
    // FIXME: This is the slow way to compare, we can use fast non-generic 
    //        implementations once the query engine is working.
    this.valueComparator = columnTypeTools.getGenericComparator();
  }

  public IDrawerIndexReader<VT> getDrawerHeader() {
    return this.indexReader;
  }

  public IColumnSegmentInfo<VT> getSegmentInfo() {
    return this.drawerInfo;
  }

  /** @see IColumnSegmentReader#hasEntryIndex(long) */
  public boolean hasEntryIndex(final long targetEntryIndex) {
    if (this.drawerInfo == null) { return false; }
    long firstIndex = this.drawerInfo.getFirstEntryIndex();
    long numEntries = this.drawerInfo.getEntryCount();

    return targetEntryIndex >= firstIndex
        && targetEntryIndex < (firstIndex + numEntries);
  }

  /** @see IColumnSegmentReader#jumpCursor(int, IValueReference<VT>) */
  public void jumpCursor(final int targetEntryIndex,
                         final IValueReference<VT> valueOut) {
    assert valueOut != null;

    if (!this.dataCardReader.hasEntryIndex(targetEntryIndex)) {
      boolean hasCard =
          this.indexReader.getCardDataForEntryIndex(targetEntryIndex, this.cardInfo);
      assert hasCard; // jumpCursor should not be called for invalid indexes
      this.dataCardReader.resetCard(this.cardInfo);
    }

    this.dataCardReader.jumpCursor(targetEntryIndex, valueOut);
  }

  /**
   * @see IColumnSegmentReader#jumpCursorToValue(IValueReference<VT>,
   *      IValueReference<VT>)
   */
  public boolean jumpCursorToValue(final IValueReference<VT> minimumTargetValue,
                                   final IValueReference<VT> valueOut) {
    assert valueOut != null;
    
    if (!this.dataCardReader.hasValue(minimumTargetValue)) {
      boolean hasCard = 
          this.indexReader.getCardDataForValue(minimumTargetValue, this.cardInfo);
      if (!hasCard) {
        return false;  // Target is beyond domain of this column.
      }
      assert this.cardInfo.getCardIndex() >= 0;
      this.dataCardReader.resetCard(this.cardInfo);
    }
    
    return this.dataCardReader.jumpCursorToValue(minimumTargetValue, valueOut);
  }

  public void resetDrawer(final ByteBuffer drawerBuffer,
                          final ColumnSegmentInfo<VT> drawerInfo) {
    this.drawerInfo = drawerInfo;
    this.dataCardReader.resetDrawer(drawerBuffer, drawerInfo);
    this.indexReader.resetDrawer(drawerBuffer, drawerInfo);
  }

  /** @see IColumnSegmentReader#stepCursor(IValueReference<VT>) */
  public boolean stepCursor(final IValueReference<VT> valueOut) {
    assert valueOut != null;

    if (this.dataCardReader.stepCursor(valueOut)) {
      return true;
    } else if (this.indexReader.getNextCardData(this.cardInfo)) {
      this.dataCardReader.resetCard(this.cardInfo);
      boolean stepped = this.dataCardReader.stepCursor(valueOut);
      assert stepped;
      return true;
    } else {
      return false;
    }
  }

  @Override
  public void close() {
    // TODO Auto-generated method stub
  }

  @Override
  public boolean hasValue(IValueReference<VT> targetValue) {
    if (this.drawerInfo == null) { return false; }
    
    IValueReference<VT> firstValue = this.drawerInfo.getFirstValue();
    IValueReference<VT> lastValue = this.drawerInfo.getLastValue();
    
//    LOG.info(firstValue.ToString());
//    LOG.info(lastValue.ToString());
    
    return this.valueComparator.compare(targetValue, firstValue) >= 0
        && this.valueComparator.compare(targetValue, lastValue) <= 0;
  }

  @Override
  public boolean isInitialized() {
    return this.initialized;
  }

  public long currentIndex() {
    return this.dataCardReader.currentIndex();
  }
}
